<?php

namespace App\Web\Model\RabbitMQ;

//use App\Service\RabbitMQ\RabbitMQProducer;
use App\Web\Model\CaseResume as CaseResumeModel;
use App\Web\Model\Model;
use EchoPHP\Facades\Db;
use EchoPHP\Facades\Log;
use EchoPHP\Facades\Redis;
use EchoPHP\Rabbitmq\Query\QueueConsumer;
use EchoPHP\Rabbitmq\Query\QueueProducer;

class ResumeUpdate extends Model {

    /**
     * 简历更新后，将简历id记录到队列中，由es引擎获取然后同步
     * @param $resumeId
     * @return bool
     */
    private $resultAck = [];
    private $resultNack = [];

    /**
     * 检查简历是否需要更新，在RpoResume,ResumeDownload中的都需要更新
     * @param $resumeId
     */
    public function getResumeInRpo(array $resumeIds) {
        $result = ['status' => false, 'message' => '', 'data' => ''];
        if (empty($resumeIds)) {
            $result['message'] = '参数不可为空';
            return $result;
        }

        $resumeIds = array_unique($resumeIds);
        $caseResumeModel = new CaseResumeModel();
        $resumeIdsInEs = $caseResumeModel->checkResumeInElasticeSearch($resumeIds);
        return $resumeIdsInEs;

        /*
        try {
            //查找是否是下载的简历
            $downloadResumeIds = [];
            $inFlag = substr(str_repeat('?,', count($resumeIds)), 0, -1);
            $sql = 'SELECT distinct resumeid FROM ResumeDownload WHERE resumeid IN(' . $inFlag . ')';
            $var = $resumeIds;
            $result = Db::connect()->select($sql, $var);

            foreach ($result as $item) {
                $downloadResumeIds[] = intval($item['resumeid']);
            }
            $noDownResumeIds = array_diff($resumeIds, $downloadResumeIds);

            //简历id全部在已下载简历库中，全部需要更新
            if (empty($noDownResumeIds)) {
                return $resumeIds;
            }

            //判断是否在rpo简历库中
            $uploadResumeIds = [];
            $inFlag = substr(str_repeat('?,', count($noDownResumeIds)), 0, -1);
            $sql = 'SELECT resumeid FROM RpoResume WHERE resumeid IN(' . $inFlag . ') AND status=?';
            $var = array_merge($noDownResumeIds, ['01']);
            $uploadList = Db::connect()->select($sql, $var);
            foreach ($uploadList as $item) {
                $uploadResumeIds[] = intval($item['resumeid']);
            }

            $noUploadResumeIds = array_diff($noDownResumeIds, $uploadResumeIds);
            if (empty($noUploadResumeIds)) {
                return $noDownResumeIds;
            }

            //加入职位的简历
            $caseResumeIds = [];

            $redisObj = Redis::connect('resumeincase');
            $errResumeId = '';
            foreach ($noUploadResumeIds as $resumeId) {
                try {
                    $resultHash = $redisObj->scanHash('resumeincase:' . $resumeId);
                    if (!empty($resultHash) && count($resultHash) > 0) {
                        $caseResumeIds[] = intval($resumeId);
                    }
                } catch (Exception $ex) {
                    $errResumeId .= $resumeId . ',';
                }
            }

            if (!empty($errResumeId)) {
                Log::error('精英网调用简历更新RPO,redis查询出错:', [
                    'resumeId' => trim($errResumeId, ',')
                ]);
            }

        } catch (Exception $ex) {
            Log::error('精英网调用简历更新RPO,出错:[' . $ex->getMessage() . ']', [
                'resumeId' => $resumeIds
            ]);
        }

        return array_merge($downloadResumeIds, $uploadResumeIds, $caseResumeIds);
        */
    }


    /**写简历ID到队列服务器
     * @param $resumeIds
     * @return array
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    public function addToQueue($resumeIds) {
        $resultInfo = ['resultAck' => [], 'resultNack' => []];
        if (empty($resumeIds)) {
            return $resultInfo;
        }
        if (!is_array($resumeIds)) {
            $resumeIds = [$resumeIds];
        }

        $configMQ = config('rabbitmq')['businesses']['resumeupdate'];


        $useExchange = $configMQ['exchange'];  //resumeupdate
        $useQueue = $configMQ['exchange'];     //updateresume_resumelist

        //指定发消息用的交换机
        $produce = new QueueProducer($useExchange);

        $messages = [];
        foreach ($resumeIds as $resumeId) {
            $messages[] = ['message' => $resumeId, 'routekey' => 'updateresume_resumelist'];
            //$messages[] = ['message' => $resumeId, 'routekey' => 'updateresume_resumelist'];
        }

        $result = $produce->produce($messages, 'updateresume_resumelist', false, [$this, 'dealAck'], [$this, 'dealNack']);

        $resultInfo['resultAck'] = $this->resultAck;
        $resultInfo['resultNack'] = $this->resultNack;
        if (count($resultInfo['resultNack']) > 0) {
            Log::error('更新简历ID到队列服务器失败：' . json_encode($resultInfo['resultNack']));
        }
        return $resultInfo;
    }


    /**单条方式读队列
     * @return bool
     */
    public function read() {
        return;  // 开发测试用

        try {
            $consumer = new QueueConsumer('resumeupdate');   //
            $deal = function (\AMQPEnvelope $message) {
                //消费消息
                var_dump($message->getBody());
                return true;
            };

            $queueName = 'updateresume_resumelist';
            $result = $consumer->get($deal, $queueName, true, false);
        } catch (\Exception $e) {
            Log::error('队列消费出现异常:' . $e->getMessage());
            return false;
        }
    }

    /** 阻塞方式读队列
     * @return bool
     */
    public function reading() {
        return;  // 开发测试用

        try {
            $consumer = new QueueConsumer('resumeupdate');
            $deal = function (\AMQPEnvelope $message) {
                var_dump($message->getBody());
                Log::error('队列消费:' . $message->getBody());
                return true;
            };
            $queueName = 'updateresume_resumelist';
            $consumer->consume($deal, $queueName, false);
        } catch (\Exception $e) {
            var_dump($e->getMessage());
            Log::error('队列消费出现异常:' . $e->getMessage());
            return false;
        }
    }

    /**
     * @param $delivery_tag
     * @param array $messageInfo
     * @return bool
     */
    public function dealAck($delivery_tag, $messageInfo) {
        $this->resultAck[] = ['delivery_tag' => $delivery_tag, 'message' => $messageInfo];
        return true;
    }

    /**
     * @param $delivery_tag
     * @param array $messageInfo
     * @return bool
     */
    public function dealNack($delivery_tag, $messageInfo) {
        $this->resultNack[] = ['delivery_tag' => $delivery_tag, 'message' => $messageInfo];
        return true;
    }
}